package com.hhe.datax_transfer.service;

import cn.hutool.core.io.FileUtil;
import cn.hutool.core.io.NioUtil;
import cn.hutool.core.util.StrUtil;
import com.hhe.datax_transfer.cache.DataxJobMonitoringControl;
import com.hhe.datax_transfer.cache.ExecuteJobsQueue;
import com.hhe.datax_transfer.cache.TransferData;
import com.hhe.datax_transfer.common.EnumDefine;
import com.hhe.datax_transfer.common.StringDefine;
import com.hhe.datax_transfer.component.TransferComponent;
import com.hhe.datax_transfer.exception.ServiceException;
import com.hhe.datax_transfer.model.DataXJsonExecuteModel;
import com.hhe.datax_transfer.model.DataxJsonQueryFilterModel;
import com.hhe.datax_transfer.model.DirectoryModel;
import com.hhe.datax_transfer.model.TransferDataModel;
import com.hhe.datax_transfer.utils.CollectionUtils;
import com.hhe.datax_transfer.utils.StringUtils;
import com.hhe.datax_transfer.vo.DataxJsonListVO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.servlet.ServletOutputStream;
import javax.servlet.http.HttpServletResponse;
import java.io.*;
import java.util.*;
import java.util.stream.Collectors;

/**
 * 数据迁移操作
 *
 * @author HHe
 * @version V1.0
 * @date 2021/1/8 15:35
 */
@Slf4j
@Component
public class TransferOperation {
    @Autowired
    private TransferComponent transferComponent;
    @Autowired
    private TransferData transferData;
    @Autowired
    private AsyncExecuteTransferOperation asyncExecuteTransferOperation;
    @Value("${execute.ceiling}")
    private String executeCeiling;

    /**
     * 获取配置文件列表
     *
     * @param dataxJsonQueryFilterModel 查询过滤条件
     * @return {@link List< DataxJsonListVO>}
     * @version V1.0
     * @author HHe
     * @date 2021/1/12 10:27
     */
    public List<DataxJsonListVO> getDataXJobList(DataxJsonQueryFilterModel dataxJsonQueryFilterModel) {
        //DataX文件所在文件夹
        String dataxJsonLibPath = transferComponent.getLocalProjectPath() + StringDefine.DATAX_JSON_PATH;
//        log.info("查询文件->{}", dataxJsonLibPath);
        //文件夹不存在则创建文件夹
        File dataxFile = new File(dataxJsonLibPath);
        if (!dataxFile.exists()) {
            dataxFile.mkdirs();
        }
        //路径为空并且是以目录路径开头
        if (StringUtils.isNotTrimBlank(dataxJsonQueryFilterModel.getCatalogue()) && dataxJsonQueryFilterModel.getCatalogue().startsWith(dataxJsonLibPath)) {
            dataxJsonLibPath = dataxJsonQueryFilterModel.getCatalogue();
            if (Objects.equals("1", dataxJsonQueryFilterModel.getMoveUp()) && !Objects.equals(dataxJsonLibPath, transferComponent.getLocalProjectPath() + StringDefine.DATAX_JSON_PATH)) {
                if (dataxJsonLibPath.endsWith("/")) {
                    dataxJsonLibPath = StrUtil.subBefore(dataxJsonLibPath, "/", true);
                }
                dataxJsonLibPath = StrUtil.subBefore(dataxJsonLibPath, "/", true);
            }
        }
        //获取目录下的file集合
        File catalogue = new File(dataxJsonLibPath);
        if (!catalogue.isDirectory()) {
            return new ArrayList<>();
        }
        File[] files = catalogue.listFiles();
        List<DataxJsonListVO> list = new ArrayList<>();
        for (File file : files) {
            list.add(new DataxJsonListVO(file.getName(), FileUtil.normalize(file.getPath()), file.isDirectory() ? EnumDefine.file_type.folder.toString() : EnumDefine.file_type.file.toString(), transferData.getTransferDataModelByFullPath(FileUtil.normalize(file.getPath())).getRemarks()));
        }
        //文件夹在上排序
        Collections.sort(list, (o1, o2) -> {
            //判断是否是同级
            if (Objects.equals(o1.getFileType(), o2.getFileType())) {
                return o1.getFullPathName().compareTo(o2.getFullPathName());
            } else {
                //不同级文件夹在上
                if (Objects.equals(o2.getFileType(), "folder")) {
                    return 0;
                }
                return -1;
            }
        });
        return list;
    }

    /**
     * 修改配置文件信息
     *
     * @param transferDataModel 配置信息
     * @return {@link int}
     * @version V1.0
     * @author HHe
     * @date 2021/1/12 16:07
     */
    public int updateDataXJsonMess(TransferDataModel transferDataModel) {
        transferData.updateDataMapMess(transferDataModel);
        return 0;
    }

    /**
     * 写入配置文件
     *
     * @return {@link Object}
     * @version V1.0
     * @author HHe
     * @date 2021/1/13 9:38
     */
    public int writeDataXJsonMess() {
        return transferData.writeDataXJsonMess();
    }

    /**
     * 执行DataX配置文件
     *
     * @param dataXJsonExecuteModel DataX执行Model
     * @return {@link int}
     * @version V1.0
     * @author HHe
     * @date 2021/1/13 10:48
     */
    public int executeDataXJson(DataXJsonExecuteModel dataXJsonExecuteModel) {
        List<String> runningName = DataxJobMonitoringControl.judgeHaveRunningAlsoReturn(dataXJsonExecuteModel.getFullPathNames());
        List<String> fileNameList = new ArrayList<>();
        if (!CollectionUtils.judgeNullCollection(runningName)) {
            throw new ServiceException("201", "任务【" + runningName.stream().collect(Collectors.joining("】,【")) + "】正在执行中,不可重复执行");
        }
        //验证其中每一条是否是文件，暂时只处理文件，后期可读取文件夹中所有文件执行
        for (String str : dataXJsonExecuteModel.getFullPathNames()) {
            File file = new File(str);
            if (!file.exists()) {
                throw new ServiceException("201", "文件【" + str + "】不存在");
            }
            //如果是文件夹获取文件夹中所有".json"尾缀文件的全路径名
            if (file.isDirectory()) {
                List<String> filePaths = FileUtil.loopFiles(file, (pathname -> pathname.getName().endsWith(".json"))).stream().map(s -> FileUtil.normalize(s.getPath())).collect(Collectors.toList());
                filePaths.forEach(s -> log.info(s));
                fileNameList.addAll(filePaths);
                continue;
            }
            if (str.endsWith(".json")) {
                fileNameList.add(str);
            }
        }
        if (!CollectionUtils.judgeNullCollection(fileNameList)) {
            for (String fullPathName : fileNameList) {
                if (ExecuteJobsQueue.getList().contains(fileNameList)) {
                    throw new ServiceException("201", "【" + fileNameList + "】处于执行队列中");
                }
                if (DataxJobMonitoringControl.queryRunningJobList().contains(fileNameList)) {
                    throw new ServiceException("201", "【" + fileNameList + "】正在执行，不可再次执行");
                }
                ExecuteJobsQueue.put(fullPathName);
            }
        }
        return 0;
    }

    @Scheduled(cron = "0/5 * *  * * ? ")
    public void scheduledExecuteQueueJobs() {
        if (!ExecuteJobsQueue.underwayExecute() && ExecuteJobsQueue.isEmpt() && !DataxJobMonitoringControl.runningLogsIsEmpty()) {
            writeLogsInFile();
        }
        //是否正在执行
        if (!ExecuteJobsQueue.judgeExecute() || ExecuteJobsQueue.isEmpt()) {
            return;
        }
        String jobFullPathName = ExecuteJobsQueue.get();
        log.info("定时任务开始执行文件：{}", jobFullPathName);
        asyncExecuteTransferOperation.executeDataXJson(jobFullPathName);
    }

    private void writeLogsInFile() {
        //获取路径
        String folderPath = transferComponent.getLocalProjectPath() + StringDefine.DATAX_JSON_LOGS_PATH;
        File folder = new File("folderPath");
        if (!folder.exists()) {
            folder.mkdirs();
        }
        File file = new File(folderPath + "/runninglogs.log");
        if (!file.exists()) {
            try {
                file.createNewFile();
            } catch (IOException e) {
                log.error("写入运行日志时创建文件异常，异常报告：{}", e.getMessage());
            }
        }
        //写入文件
        FileUtil.writeLines(DataxJobMonitoringControl.getRuningLogs(), file, "UTF-8");
        //清空日志
        DataxJobMonitoringControl.clearRuningLogs();
    }

    /**
     * 终止DataX执行任务
     *
     * @param transferDataModel DataX执行Model
     * @return {@link int}
     * @version V1.0
     * @author HHe
     * @date 2021/3/9 10:17
     */
    public int interruptDataXJsonJob(TransferDataModel transferDataModel) {
        //停止信息
        Thread executeJobThreadMess = DataxJobMonitoringControl.getExecuteJobThreadMess(transferDataModel.getFullPathName());
        if (executeJobThreadMess != null) {
            executeJobThreadMess.interrupt();
        }
        DataxJobMonitoringControl.executeFinish(transferDataModel.getFullPathName());
        return 0;
    }

    /**
     * DataX配置文件内容查询
     *
     * @param dataXJsonExecuteModel
     * @param response
     * @version V1.0
     * @author HHe
     * @date 2021/1/15 17:30
     */
    public void dataxJsonFileView(TransferDataModel dataXJsonExecuteModel, HttpServletResponse response) throws IOException {
        ServletOutputStream outputStream = response.getOutputStream();
        InputStream inputStream;
        try {
            inputStream = new FileInputStream(dataXJsonExecuteModel.getFullPathName());
        } catch (Exception e) {
            inputStream = new ByteArrayInputStream(e.getMessage().getBytes());
        }
        NioUtil.copyByNIO(inputStream, outputStream, 1024 << 4, null);
        inputStream.close();
        outputStream.flush();
        outputStream.close();
    }

    /**
     * DataX配置文件日志查询
     *
     * @param dataXJsonExecuteModel
     * @param response
     * @version V1.0
     * @author HHe
     * @date 2021/1/15 17:32
     */
    public void dataxJsonLogView(TransferDataModel dataXJsonExecuteModel, HttpServletResponse response) throws IOException {
        ServletOutputStream outputStream = response.getOutputStream();
        InputStream inputStream;
        try {
            inputStream = new FileInputStream(dataXJsonExecuteModel.getFullPathName().replace("/datax_json_lib/", "/datax_json_logs/").replace(".json", ".log"));
        } catch (Exception e) {
            inputStream = new ByteArrayInputStream(e.getMessage().getBytes());
        }
        NioUtil.copyByNIO(inputStream, outputStream, 1024 << 4, null);
        outputStream.flush();
        inputStream.close();
        outputStream.close();
    }

    /**
     * 返回文件列表目录
     *
     * @param dataxJsonQueryFilterModel 文件信息
     * @return {@link Object}
     * @version V1.0
     * @author HHe
     * @date 2021/1/26 11:10
     */
    public List<DirectoryModel> getJobListDirectory(DataxJsonQueryFilterModel dataxJsonQueryFilterModel) {
        List<DirectoryModel> list = new ArrayList<>();
        //DataX文件所在文件夹
        String dataxJsonLibPath = transferComponent.getLocalProjectPath() + StringDefine.DATAX_JSON_PATH;
        list.add(new DirectoryModel("根", dataxJsonLibPath));
        if (StringUtils.isTrimBlank(dataxJsonQueryFilterModel.getCatalogue())) {
            return list;
        }
        //解析字符串
        //获取文件所在根目录
        String jobDirectory = StrUtil.subAfter(dataxJsonQueryFilterModel.getCatalogue(), dataxJsonLibPath, false);
        if (Objects.equals("1", dataxJsonQueryFilterModel.getMoveUp())) {
            jobDirectory = StrUtil.subBefore(jobDirectory, "/", true);
        }
        if (jobDirectory.trim().startsWith("/")) {
            jobDirectory = StrUtil.removePrefix(jobDirectory.trim(), "/");
        }
        //如果是"/",""则不操作直接返回
        if (!Objects.equals(jobDirectory.trim(), "/") && !Objects.equals(jobDirectory.trim(), "")) {
            StringJoiner joiner = new StringJoiner("/", dataxJsonLibPath + "/", "");
            for (String s : jobDirectory.trim().split("/")) {
                joiner.add(s);
                list.add(new DirectoryModel(s, joiner.toString()));
            }
        }
        return list;
    }

    /**
     * DataX配置文件执行状态
     *
     * @param transferDataModel 信息
     * @return {@link boolean}
     * @version V1.0
     * @author HHe
     * @date 2021/1/27 10:05
     */
    public boolean dataxJsonRunStatus(TransferDataModel transferDataModel) {
        return DataxJobMonitoringControl.haveRunningJobByName(transferDataModel.getFullPathName());
    }

    /**
     * 删除执行队列
     *
     * @param transferDataModel
     * @return {@link int}
     * @version V1.0
     * @author HHe
     * @date 2021/3/19 16:00
     */
    public int deleteExecuteQueue(TransferDataModel transferDataModel) {
        if (ExecuteJobsQueue.delete(transferDataModel.getFullPathName())) {
            return 0;
        } else {
            throw new ServiceException("201", "删除失败");
        }
    }
}
